package hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

public class HbaseDX {

    Connection conn;
    Admin admin;
    Table dx;

    @Before
    public void init() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master,node1,node2");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        dx = conn.getTable(TableName.valueOf("DIANXIN"));

    }

    @Test
    public void create() throws IOException {
        HTableDescriptor dianxin = new HTableDescriptor(TableName.valueOf("DIANXIN"));
        HColumnDescriptor cf1 = new HColumnDescriptor("cf1");
        cf1.setMaxVersions(3);
        dianxin.addFamily(cf1);
        admin.createTable(dianxin);
    }

    @Test
    public void putAll() throws IOException {
        BufferedReader br = new BufferedReader(new FileReader("D:/ALanzhishujia/soft/data/DIANXIN.csv"));
        ArrayList<Put> puts = new ArrayList<Put>();
        String line=null;
        while ((line=br.readLine())!=null){
            String[] split = line.split(",");
            String adm=split[0];
            String time = split[1];
            String jindu = split[4];
            String weidu = split[5];
            Put put = new Put(adm.getBytes());

            //将时间作为版本号
            long start_time = Long.parseLong(time);

            //put的时候，同时put上value的时间戳版本
            put.addColumn("cf1".getBytes(),"jindu".getBytes(),start_time,jindu.getBytes());
            put.addColumn("cf1".getBytes(),"weidu".getBytes(),start_time,weidu.getBytes());
            puts.add(put);

            if (puts.size()==1000){
                dx.put(puts);
                puts=new ArrayList<Put>();
            }
        }
        if (puts.size()!=0){
            dx.put(puts);
        }


    }

    @Test
    public void getthree() throws IOException {
        // hbase shell中的命令
        // hbase(main):001:0> get 'DIANXIN',
        // 'FDA42AB4AA84D13C2AC7512D1BAEE93B7DCC327A',{COLUMN=>['cf1:jindu','cf1:weidu'],VERSIONS=>3}

        String adm="FDA42AB4AA84D13C2AC7512D1BAEE93B7DCC327A";
        Get get = new Get(adm.getBytes());
        get.setMaxVersions(3);

        Result rs = dx.get(get);
        ArrayList<String> jinlist = new ArrayList<String>();
        ArrayList<String> weilist = new ArrayList<String>();
        for (Cell cell : rs.listCells()) {
            String quo = Bytes.toString(CellUtil.cloneQualifier(cell));
            String values = Bytes.toString(CellUtil.cloneValue(cell));
            if ("jindu".equals(quo)){
                jinlist.add(values);
            }else if ("weidu".equals(quo)){
                weilist.add(values);
            }
        }

        for (int i = 0; i < 3; i++) {
            System.out.println(jinlist.get(i)+","+weilist.get(i));
        }

    }

    @After
    public void close() throws IOException {
        conn.close();
    }
}
